package base

import java.util.concurrent.*

/**
 *  PipeWriter receive the cache from the invoker thread and write to the file/database in the background thread.
 *
 *  @param cacheNumberSize the number of cache in the queue
 *  @param stream the stream to write the cache
 */
class PipeWriter<T : Any>(private val cacheNumberSize: Int, private val stream: IPipeStream<T>) {

    companion object {
        const val TAG = "PipeWriter"
    }

    private val mQueue: BlockingQueue<T> = LinkedBlockingQueue()

    @Volatile
    private var mIsRunning = false

    private val mExecutorService = ThreadPoolExecutor(
        1, 1, 0L, TimeUnit.MILLISECONDS, LinkedBlockingQueue<Runnable>(),
        PipeThreadFactory(),
        ThreadPoolExecutor.AbortPolicy()
    )

    /**
     * add element to cache
     */
    fun addElement(t: T) {
        mQueue.put(t)
    }

    /**
     * start the cache thread and write to the file/database
     */
    fun start() {
        println("$TAG start --")
        mQueue.clear()
        stream.open()
        mIsRunning = true
        mExecutorService.submit(executor)
    }

    private val executor = Runnable {

        run {
            val buffer = mutableListOf<T>()

            println("$TAG executor running :${mIsRunning} --")

            while (mIsRunning) {
                try {
                    // block when the queue is empty
                    val element = mQueue.take()
                    buffer.add(element)
                    // when buffer is full,then flush
                    if (buffer.size >= cacheNumberSize) {
                        println("$TAG cache is full,flush  --")
                        stream.write(buffer)
                        buffer.clear()
                    }
                } catch (e: InterruptedException) {
                    Thread.currentThread().interrupt()
                }
            }

            println("$TAG mIsRunning break --")

            //check if the queue is not empty , flush them
            buffer.isNotEmpty().let {
                println("$TAG cache left some element, flush --")
                stream.write(buffer)
            }

            stream.close()
        }
    }

    /**
     * stop the cache thread
     */
    fun stop() {
        println("$TAG stop --")

        mIsRunning = false
        mExecutorService.shutdownNow()
    }
}